package com.flink.timewindow.window;

import com.flink.timewindow.bean.WaterSensor;
import com.flink.timewindow.function.WaterSensorMapFunction;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class WindowApiDemo {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据源
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("10.90.100.102", 8888)
                //数据处理
                //切分转换
                .map(new WaterSensorMapFunction());
        //分组
        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(value -> value.getId());

        //窗口设置
        //TODO 1. 指定窗口分配器： 指定窗口类型
           //1.1 没有keyby的窗口：窗口内的 所有数据 进入同一个 子任务，并行度只能为1
           //sensorDS.windowAll();
           //1.2 有keyby的窗口： 每个key上都定义一个组窗口 ，各自独立的进行统计计算

           //如 基于时间的
           sensorKS.window(TumblingProcessingTimeWindows.of(Time.seconds(10)));//滚动窗口 窗口长度10s
           //sensorKS.window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5)));//滑动窗口 窗口长度10s 滑动步长5s
           //sensorKS.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)));//会话窗口，超过间隔5s就是新的窗口

          //基于计数
          sensorKS.countWindow(5);//滚动窗口，窗口长度为5个元素
          //sensorKS.countWindow(5,2);//滑动窗口，窗口长度为5个元素，滑动步长为2个元素
          //全局窗口，计数窗口的底层就是用的这个，需要自定义触发器（可以看源码），需要自定义的时候才会用
          sensorKS.window(GlobalWindows.create());

        //TODO 2. 指定窗口函数：窗口内数据的计算逻辑




        //执行
        env.execute();

    }
}
